﻿using MongoDB.Bson;
using MongoDB.Driver;
using System;
using System.Collections.Generic;
using System.Linq;
using System.Text;
using System.Threading.Tasks;
using System.Data;
using NLog;
using System.Threading;
using System.Diagnostics;
using MySql.Data.MySqlClient;
using DevLib.Data;
using System.Configuration;
using System.Globalization;
using System.Linq.Expressions;

namespace sqltomongo
{

    public class ParamllelTask
    {
        private ParamllelTask(){}

        private static Logger logger = LogManager.GetCurrentClassLogger();
        private static SQLServerDb sqlDb = SQLServerDb.Instance();
        private static IMongoCollection<BsonDocument> mongoCollection;
        /// <summary>
        /// 任务开始
        /// </summary>
        /// <param name="pageSize"></param>
        /// <param name="channel"></param>
        public static void Start(int pageSize, string channel)
        {
            int totalSize;
            int totalPage = GetPageSize(pageSize, channel, out totalSize);
            var message = $"---即将同步分区【{channel}】,分区记录总数量:{totalSize},每页:{pageSize}条,共计:{totalPage}页---" + Environment.NewLine;
            WriteLog(message);
            for (int i = 1; i <= totalPage; i++)
            {
                PaggingQuery(pageSize, i, channel: channel);
            }
        }
        /// <summary>
        /// 分页并行查询（暂未启用）
        /// </summary>
        /// <param name="perCount"></param>
        /// <param name="channel"></param>
        public static void StartParellelThread(int perCount, string channel)
        {
            Func<int, int, int, string, bool> execQuery = (pageSize, pageNumber, skipSize, thread) =>
            {
                int totalSize;
                int totalPage = GetPageSize(pageSize, pageNumber, skipSize, channel, out totalSize);
                var message = $"---【{thread}】记录总数量:{totalSize},每页:{pageSize}条,共计:{totalPage}页---" + Environment.NewLine;
                WriteLog(message);
                for (int i = skipSize; i <= pageNumber; i++)
                {
                    PaggingQuery(pageSize, i, notice: thread, channel: channel);
                }
                return true;
            };
            Parallel.Invoke(() => execQuery(perCount, 30, 1, "线程-1"), () => execQuery(perCount, 61, 31, "线程-2"));
        }
        /// <summary>
        /// 分页查询同步到mongodb
        /// </summary>
        /// <param name="pageSize"></param>
        /// <param name="pageNumber"></param>
        /// <param name="notice"></param>
        /// <param name="channel"></param>
        public static void PaggingQuery(int pageSize, int pageNumber, string notice = "", string channel = "")
        {
            string ennableChannel = !string.IsNullOrEmpty(channel) ? $" where Channel={channel}" : "";
            var querySQL = string.Format(SQL, pageSize, ennableChannel, pageNumber);
            var startMsg = $"---start：  {notice}同步分区【{channel}】，第【{pageNumber}】页，数据开始---";
            WriteLog(startMsg);
            try
            {
                ListOrCallbackData(querySQL,listBsonDocuments=>
                {
                    MongoCollection.InsertMany(listBsonDocuments, new InsertManyOptions { BypassDocumentValidation = false });
                    var endMsg = $"---finish： {notice}同步分区【{channel}】，第【{pageNumber}】页，数据完成---\r\n\r\n";
                    WriteLog(endMsg);
                    listBsonDocuments.Clear();
                    return false;
                });
            }
            catch (Exception ee)
            {
                var errMsg = $"---{notice}同步异常,原因->{ee.Message}---";
                Console.WriteLine(errMsg);
                logger.Error(errMsg);
            }
        }
        /// <summary>
        /// 获取分页数
        /// </summary>
        /// <param name="pageSize"></param>
        /// <param name="startNumber"></param>
        /// <param name="skipSize"></param>
        /// <param name="channel"></param>
        /// <param name="total"></param>
        /// <returns></returns>
        private static int GetPageSize(int pageSize, int startNumber, int skipSize, string channel, out int total)
        {
            string ennableChannel = !string.IsNullOrEmpty(channel) ? $" where Channel={channel}" : "";
            string sql_thread = $@"SELECT count(id) 
            FROM 
	            (
	                SELECT ROW_NUMBER() OVER (ORDER BY id) AS RowNumber,* FROM Fct_Customer  {ennableChannel}
	            )   as A  WHERE
             " + (skipSize <= 1 ? $" RowNumber<={pageSize}*{startNumber}" : $" RowNumber>{pageSize}*{skipSize} and RowNumber<={pageSize}*{startNumber}");
            total = 0;
            try
            {
                int totalSize = (int)sqlDb.ExecuteScalar(CommandType.Text, sql_thread, null);
                total = totalSize;
                decimal pages = Math.Ceiling(decimal.Parse(totalSize.ToString()) / decimal.Parse(pageSize.ToString()));
                return (int)pages;
            }
            catch (Exception ee)
            {
                logger.Error($"获取总页数失败:{ee.Message}");
            }
            return 0;
        }
        public  static void UpSertData()
        {
            try
            {
                List<BsonDocument> listBsonDocuments = ListOrCallbackData(SqlExternel);
                WriteLog($"SQL时间区间内待处理数据:{listBsonDocuments.Count}条");
                var timeSpan = new TimeSpan(0, 0, 9999);
                if (listBsonDocuments.Count > 0)
                {

                    string[] idArray = listBsonDocuments.Select(item => item["ID"].ToString().ToLower()).ToArray();
                    var bsonDocument = new BsonDocument() { { "ID", new BsonDocument { { "$in", new BsonArray(idArray) } } } };
                    FindOptions options = new FindOptions { MaxTime = timeSpan, MaxAwaitTime = timeSpan, NoCursorTimeout = true, AllowPartialResults = true };
                    var projection = Builders<BsonDocument>.Projection.Exclude("_id");
                    List<BsonDocument> findDocuments = MongoCollection.Find(bsonDocument, options).Project(projection).ToList();
                    var findDocumentsString = string.Join(",", findDocuments.Select(item => item["ID"]+Environment.NewLine));
                    //WriteLog($"删除记录ID明细:{findDocumentsString}");
                    logger.Info($"删除记录ID明细:\r\n{findDocumentsString}");
                    DeleteResult deleteResult = mongoCollection.DeleteMany(bsonDocument, new DeleteOptions() { Collation = Collation.Simple });
                    long deletedCount = deleteResult.DeletedCount;
                    WriteLog($"mongodb中删除记录{deletedCount}条");
                    mongoCollection.InsertManyAsync(listBsonDocuments, new InsertManyOptions() { BypassDocumentValidation = false });
                    WriteLog($"mongodb中插入记录{listBsonDocuments.Count}条");
                    var insertDocumentsString = string.Join(",", listBsonDocuments.Select(item => item["ID"]+ Environment.NewLine));
                    logger.Info($"插入记录ID明细:\r\n{insertDocumentsString}");





                    //string[] idArray = listBsonDocuments.Select(item => item["ID"].ToString().ToLower()).ToArray();
                    //var bsonDocument = new BsonDocument() { { "ID", new BsonDocument { { "$in", new BsonArray(idArray) } } } };
                    //FindOptions options = new FindOptions { MaxTime = timeSpan, MaxAwaitTime = timeSpan, NoCursorTimeout=true, AllowPartialResults=true };
                    //var projection = Builders<BsonDocument>.Projection.Exclude("_id");
                    //var findResult = MongoCollection.Find(bsonDocument, options).Project(projection);
                    //var excepResutList=new List<BsonDocument>();
                    //List<BsonDocument> resultList = null;
                    //if (findResult.Count() > 0)
                    //{
                    //    excepResutList.AddRange(findResult.ToList());
                    //    resultList= findResult.ToList();
                    //    WriteLog($"mongodb中检索到记录{resultList.Count}条");
                    //    var resultArray = (BsonDocument[])resultList.ToArray().Clone();

                    //    DeleteResult deleteResult = mongoCollection.DeleteMany(bsonDocument, new DeleteOptions() { Collation = Collation.Simple });
                    //    long deletedCount = deleteResult.DeletedCount;
                    //    if (deletedCount > 0)
                    //    {
                    //        WriteLog($"mongodb中删除记录{deletedCount}条");
                    //        mongoCollection.InsertManyAsync(resultArray, new InsertManyOptions() { BypassDocumentValidation = false });
                    //        WriteLog($"mongodb中插入记录{resultArray.Length}条");
                    //    }
                    //}
                    //if (excepResutList.Count > 0)
                    //{
                    //    excepResutList.ForEach(item =>
                    //    {
                    //        var document=listBsonDocuments.Find(bson => bson["ID"] == item["ID"]);
                    //        listBsonDocuments.Remove(document);
                    //    });
                    //}
                    //mongoCollection.InsertManyAsync(listBsonDocuments, new InsertManyOptions() { BypassDocumentValidation = false });
                    //WriteLog($"mongodb中新增记录{listBsonDocuments.Count}条");


                    //listBsonDocuments = listBsonDocuments.Except(excepResutList).ToList();
                    //if (listBsonDocuments.Count() > 0)
                    //{
                    //    mongoCollection.InsertManyAsync(listBsonDocuments, new InsertManyOptions() { BypassDocumentValidation = false });
                    //    WriteLog($"mongodb中新增记录{listBsonDocuments.Count}条");
                    //}
                }
            }
            catch (Exception ee)
            {
                WriteLog($"系统异常:{ee.Message}");
                throw new Exception(ee.Message);
            }
        }
        public static void WriteLog(string logMsg) {
            Console.WriteLine(logMsg);
            logger.Info(logMsg);
        }
        private static List<BsonDocument> ListOrCallbackData(string querySQL, Func<List<BsonDocument>,bool> expression=null)
        {
            List<BsonDocument> listBsonDocument = new List<BsonDocument>();
            DataSet ds = SQLServerDb.IsSQLServer ? sqlDb.ExecuteSQLDataSet(querySQL) : MySqlHelper.ExecuteDataSet(querySQL);
            if (ds.Tables[0].Rows.Count > 0)
            {
                var captions = ds.Tables[0].Columns;
                foreach (DataRow row in ds.Tables[0].Rows)
                {
                    var bsonDocument = new BsonDocument();
                    for (int i = 0; i < captions.Count; i++)
                    {
                        var columnName = captions[i].ToString();
                        bsonDocument[columnName] = row[columnName] == null ? "" : row[columnName].ToString().Trim();
                    }
                    listBsonDocument.Add(bsonDocument);
                }
                expression?.Invoke(listBsonDocument);
            }
            return listBsonDocument;
        }
        /// <summary>
        /// 获取分页数
        /// </summary>
        /// <param name="pageSize"></param>
        /// <param name="channel"></param>
        /// <param name="total"></param>
        /// <returns></returns>
        private static int GetPageSize(int pageSize, string channel, out int total)
        {
            total = 0;
            try
            {
                string sql = "select count(id) from Fct_Customer";
                sql += (!string.IsNullOrEmpty(channel) ? $" where channel={channel}" : "");
                int totalSize = (int)sqlDb.ExecuteScalar(CommandType.Text, sql, null);
                total = totalSize;
                decimal pages = Math.Ceiling(decimal.Parse(totalSize.ToString()) / decimal.Parse(pageSize.ToString()));
                return (int)pages;
            }
            catch (Exception ee)
            {
                logger.Error($"获取总页数失败:{ee.Message}");
            }
            return 0;
        }
        private static object obj = new object();
        private static IMongoCollection<BsonDocument> MongoCollection {
            get
            {
                if (mongoCollection == null)
                {
                    lock (obj)
                    {
                        MongoDB<BsonDocument> mongoDB = new MongoDB<BsonDocument>();
                        mongoCollection = mongoDB.GetCollection(TargetCollection);
                    }
                }
                return mongoCollection;
            }
        }
        private static string GetConfigString(string keyName)
        {
            return !string.IsNullOrEmpty(keyName) ? ConfigurationManager.AppSettings[keyName].ToString().Trim() : "";
        }
        private static string SQL { get; set; } = GetConfigString("sql_statement");
        private static string SqlExternel { get; } = GetConfigString("sql_externel");
        private static string TargetCollection { get; } = GetConfigString("collection_target");
    }
}